package com.sink;

import com.dto.ProjectBaseInfo;
import com.utils.NumberUtils;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import com.alibaba.fastjson.JSONObject;

import java.time.LocalDateTime;

/**
 * @description:
 * @author: sksun2
 * @create: 2023-08-31
 */
public class ASink extends RichSinkFunction<String> {

    @Override
    public void invoke(String value, Context context) throws Exception {
        System.out.println("监听到活动数据" + LocalDateTime.now() + value);


        JSONObject jsonObject = JSONObject.parseObject(value);

        ProjectBaseInfo before = jsonObject.getObject("before", ProjectBaseInfo.class);
        ProjectBaseInfo after = jsonObject.getObject("after", ProjectBaseInfo.class);

        initNum(after,jsonObject.getJSONObject("after").getJSONObject("PROJECT_SOURCE"));

        System.out.println();

    }

    private void initNum(ProjectBaseInfo projectBaseInfo,JSONObject jsonObject){
        final Integer scale = jsonObject.getInteger("scale");
        final String value = jsonObject.getString("value");
        projectBaseInfo.setPROJECT_SOURCE(NumberUtils.getNumber(scale,value));
    }

}
